
package com.huaweicloud.haydn.delivercore.agent.common.consistency;

import java.text.MessageFormat;
import java.util.List;
import java.util.stream.Collectors;

import org.apache.commons.lang3.StringUtils;

import com.google.common.base.Joiner;
import com.huaweicloud.haydn.delivercore.agent.common.DbTypeEnum;
import com.huaweicloud.haydn.delivercore.agent.common.exception.BusinessErrorEnum;
import com.huaweicloud.haydn.delivercore.agent.common.exception.BusinessException;
import com.huaweicloud.haydn.delivercore.agent.dto.haydn.ContentConsistencyConfigDto;
import com.huaweicloud.haydn.delivercore.agent.dto.haydn.MonitoringFieldMappingConfigDto;

import lombok.AllArgsConstructor;
import lombok.Getter;
import lombok.NoArgsConstructor;
import lombok.extern.slf4j.Slf4j;

/**
 * 功能描述: 用于一致性对比时的源端和目的端sql的获取
 *
 * @since 2023-03-17
 */
@NoArgsConstructor
@AllArgsConstructor
@Getter
@Slf4j
public enum ConsistencySqlDefinition {

    MYSQL(DbTypeEnum.MYSQL, "Select {0} from {1}.{2} {3} ORDER BY rand() LIMIT {4}",
        "Select {0} from {1}.{2} where {3}"),
    // rownum都是从1开始
    ORACLE(DbTypeEnum.ORACLE,
        "select * from (select {0},rownum r from \"{1}\".\"{2}\" {3}) where r >= {4} and r <= {5}",
        "Select {0} from \"{1}\".\"{2}\" where {3}"),
    MRS_SPARK(DbTypeEnum.MRS_HUDI, "", "select {0} from {1} where {2}"),
    DWS(DbTypeEnum.DWS, "", "select {0} from {1}.{2} where {3}");

    private DbTypeEnum dbTypeEnum;

    // 当该数据库库类型数据作为源端数据的批量查询sql
    private String contentSourceSql;

    private String contentTargetSql;

    /**
     * 功能描述：用于生成一致性比对的sql
     *
     * @param round 当前处于抽取的轮次（大于1，小于传入batch)
     * @param configDto 表内容一致性比对使用的配置
     * @param tableName 抽取的表名称
     * @param dbName 数据库名称
     * @param schemaName shcema名称
     * @since 2023-03-20
     */
    public static String getSourceContentSql(String tableName, int round, ContentConsistencyConfigDto configDto,
        String dbName, String schemaName) {
        // 获取对应的要查询列的列表，如果没有主键，则查询所有的数据
        List<String> queryFields = configDto.getMonitoringFieldMappings()
            .stream()
            .map(mapping -> mapping.getSourceFieldNameEn())
            .collect(Collectors.toList());

        // 用逗号间隔开的、要查询的所有的字段名
        String queryFieldJoinStr = Joiner.on(",").skipNulls().join(queryFields);
        // 获取过滤sql
        String filterStatement = StringUtils.isEmpty(configDto.getFilterSqlStatement()) ? ""
            : "where ".concat(configDto.getFilterSqlStatement().replaceAll("where", ""));

        String sql;
        switch (configDto.getSourceDbType()) {
            case MYSQL:
                sql = MessageFormat.format(MYSQL.getContentSourceSql(), queryFieldJoinStr, dbName, tableName,
                    filterStatement, configDto.getSingleExtractSize());
                break;
            case ORACLE:
                int startIndex = (round - 1) * configDto.getSingleExtractSize() + 1;
                int endIndex = round * configDto.getSingleExtractSize();
                sql = MessageFormat.format(ORACLE.getContentSourceSql(), queryFieldJoinStr, schemaName, tableName,
                    filterStatement, startIndex, endIndex);
                break;
            case DWS:
            case MRS_HUDI:
            default:
                log.error("Unsupported database type : {}!", configDto.getSourceDbType());
                throw new BusinessException(BusinessErrorEnum.BASE_BAD_REQUEST, "Unsupported database type!");
        }

        return sql;
    }

    /**
     * 功能描述：用于生成一致性比对的sql
     *
     * @param configDto 表内容一致性比对使用的配置
     * @param tableName 抽取的表名称
     * @param dbName 数据库名称
     * @param schemaName shcema名称
     * @since 2023-03-20
     */
    public static String getTargetContentSql(String tableName, ContentConsistencyConfigDto configDto, String dbName,
        String schemaName, List<String> perLineSourceData) {
        // 获取对应的要查询列的列表
        List<String> queryFields = configDto.getMonitoringFieldMappings()
            .stream()
            .map(MonitoringFieldMappingConfigDto::getTargetFieldNameEn)
            .collect(Collectors.toList());

        // 用逗号间隔开的、要查询的所有的字段名
        String queryFieldJoinStr = Joiner.on(",").skipNulls().join(queryFields);

        // 获取预编译语句sql所用过滤
        String matchFilterSql = prepareTargetFilterSqlFromSource(configDto, perLineSourceData);

        String sql = "";
        switch (configDto.getTargetDbType()) {
            case DWS:
                sql = MessageFormat.format(DWS.getContentTargetSql(), queryFieldJoinStr, schemaName, tableName,
                    matchFilterSql);
                break;
            case MRS_HUDI:
                sql =
                    MessageFormat.format(MRS_SPARK.getContentTargetSql(), queryFieldJoinStr, tableName, matchFilterSql);
                break;
            case MYSQL:
                sql = MessageFormat.format(MYSQL.getContentSourceSql(), queryFieldJoinStr, dbName, tableName,
                    matchFilterSql);
                break;
            case ORACLE:
                sql = MessageFormat.format(ORACLE.getContentSourceSql(), queryFieldJoinStr, schemaName, tableName,
                    matchFilterSql);
                break;
            default:
                log.error("Unsupported database type : {}!", configDto.getTargetDbType());
                throw new BusinessException(BusinessErrorEnum.BASE_BAD_REQUEST, "Unsupported database type!");
        }

        return sql;
    }

    /**
     * 功能描述 : 获取过滤字段值的where后过滤条件，用于生出预编译语句使用
     *
     * @param configDto 表内容一致性比对使用的配置
     * @since 2023-03-21
     */
    private static String prepareTargetFilterSqlFromSource(ContentConsistencyConfigDto configDto,
        List<String> perLineSourceData) {
        List<String> targetFields = configDto.getMonitoringFieldMappings()
            .stream()
            .map(MonitoringFieldMappingConfigDto::getTargetFieldNameEn)
            .collect(Collectors.toList());
        StringBuilder findDataFilter = new StringBuilder();

        for (int i = 0; i < targetFields.size(); i++) {
            String targetField = targetFields.get(i);

            // 第一个条件紧跟where子句之后，无需添加and
            if (i != 0) {
                findDataFilter.append(" and");
            }

            if (perLineSourceData.get(i) == null) {
                findDataFilter.append(String.format(" %s is null", targetField));
            } else {
                findDataFilter.append(String.format(" %s = ?", targetField));
            }
        }

        return findDataFilter.toString();
    }

}
